package com.nhncorp.pleiades.shell;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.pleiades.client.ClientConstants;
import com.nhncorp.pleiades.client.PleiadesUser;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.hadoop.New_RPC;
import com.nhncorp.pleiades.protocol.AdminLockInfo;
import com.nhncorp.pleiades.protocol.LockAdminProtocol;
import com.nhncorp.pleiades.protocol.LockMetaData;
import com.nhncorp.pleiades.util.NetworkUtil;
import com.nhncorp.pleiades.util.Sleeper;

public class LockAdminClient implements LockAdminService, Runnable{
  
  public static final Log LOG = LogFactory.getLog(LockAdminClient.class
      .getName());  
  
  LockAdminProtocol lockAdmin = null;
  protected PleiadesConfiguration conf;
  PleiadesUser userInfo = null;
  private Sleeper sleeper = null;
  protected final AtomicBoolean stopRequested = new AtomicBoolean(false);
  
  public LockAdminClient(PleiadesConfiguration conf) throws IOException {
    try {
      this.conf = conf;
      this.userInfo = new PleiadesUser("default");
      init();
    } catch (Exception e) {
      e.printStackTrace();
      throw new IOException(e.getMessage());
    }
  }
  
  public LockAdminClient(PleiadesUser user, PleiadesConfiguration conf) throws IOException {
    try {
      this.conf = conf;
      this.userInfo = user;
      init();
    } catch (Exception e) {
      e.printStackTrace();
      throw new IOException(e.getMessage());
    }
  }  
  
  private void init() throws IOException {
    String tempName = this.userInfo.getHostName();
    this.userInfo.setHostName("adm:" + tempName);
    System.setProperty("java.net.preferIPv4Stack", "true");    
    try {
      String[] allPleiades = conf.getStrings("pleiades.cell.members");
      if(allPleiades == null || allPleiades.length == 0) {
        LOG.error("Can't get pleiades cell info check pleiades.cell.members option in pleiades-site.xml");
        throw new IOException("Can't get pleiades cell info check pleiades.cell.member option in pleiades-site.xml");
      }
      lockAdmin = getMasterPleiadesProxy(allPleiades);
      if(lockAdmin == null) {
        LOG.error("Can't get master pleiades proxy");
        throw new IOException("Can't get master pleiades proxy, check pleiades.cell.member option in pleiades-site.xml");        
      }
    } catch (Exception e) {
      throw new IOException(e.getMessage());
    }
  }
  
  private LockAdminProtocol getMasterPleiadesProxy(String[] allPleiades) throws IOException {
    
    int maxRetryWhenException = 5;
    int retry = 0;
    LockAdminProtocol pleiades = null;
    
    if(allPleiades != null && allPleiades.length > 0) {
      
      for(String entry : allPleiades) {
        LOG.debug("try to update master within servers of pleiades cell: " + entry);
        InetAddress pleiadesAddress = InetAddress.getByName(entry);
        if(!isReachable(pleiadesAddress, 8000, 5000)) {
          LOG.info(pleiadesAddress + " is not reachable");
          continue;
        }
        
        InetSocketAddress address = NetworkUtil.getAddress(entry + ":" + conf.get("clusterManager.port", String
            .valueOf(ClientConstants.DEFAULT_LOCK_PORT)));
        retry = 0;
        while (retry < maxRetryWhenException) {
          try {
            pleiades = (LockAdminProtocol) New_RPC.getProxy(
                LockAdminProtocol.class, LockAdminProtocol.versionID, address, conf);
            if(pleiades.isPrimaryPleiades()) {
              return pleiades;
            } else {
              LOG.info(entry + " is not master of primary partition");
              pleiades = null;
            }
            break;
          } catch (Exception e) {
            LOG.warn("Error while trying to update master:" + e.getMessage() + ", but retry[" + retry + "]" , e);
            pleiades = null;
            retry++;
            if (retry >= maxRetryWhenException) {
              LOG.error("Error while trying to update master:" + e.getMessage(), e);
              break;
            }
            sleeper.sleepMilliSeconds(100);
          } finally {
          }
        }
      }      
    } 
    
    return null;
  }  
  
  private boolean isReachable(InetAddress entry, int port, int timeout) {
    Socket s = new Socket();
    try {
      s.setSoTimeout(timeout);
      s.connect(new InetSocketAddress(entry, port));
      s.close();
    } catch (Exception e) {
      return false;
    }
    return true;
  }  

  
  public boolean create(String serviceId, String owner, String path,
      boolean overwrite) throws IOException {
    return lockAdmin.admin_create(new AdminLockInfo(serviceId, owner, path),
        overwrite);
  }

  
  public boolean delete(String serviceId, String path, boolean recursive)
      throws IOException {
    return lockAdmin.admin_delete(new AdminLockInfo(serviceId, userInfo
        .getHostName(), path), recursive);
  }
  
  public boolean set(String serviceId, String owner, String path,
      String key) throws IOException {
    return lockAdmin.admin_set(new AdminLockInfo(serviceId, owner, path), key);
  }

  
  public LockMetaData getMeta(String serviceId, String lockId)
      throws IOException {
    return lockAdmin.admin_getMeta(new AdminLockInfo(serviceId, userInfo
        .getHostName(), lockId));
  }

  
  public LockMetaData[] getMetaList(String serviceId, String lockId)
      throws IOException {
    return lockAdmin.admin_getMetaList(new AdminLockInfo(serviceId, userInfo
        .getHostName(), lockId));
  }

  
  public LockMetaData[] listAll(String serviceId)
      throws IOException {
    return lockAdmin.admin_listAll(new AdminLockInfo(serviceId, userInfo
        .getHostName(), ""));
  }

  
  public LockMetaData[] listById(String serviceId, String lockId)
      throws IOException {
    return lockAdmin.admin_listById(new AdminLockInfo(serviceId, userInfo
        .getHostName(), lockId));
  }

  
  public LockMetaData[] listByOwner(String serviceId, String owner)
      throws IOException {
    return lockAdmin.admin_listByOwner(new AdminLockInfo(serviceId, owner, ""));
  }

  
  public String[] listFs(String serviceId, String path) throws IOException {
    return lockAdmin.admin_listPath(new AdminLockInfo(serviceId, userInfo
        .getHostName(), path));
  }

  
  public boolean lock(String serviceId, String owner, String lockId)
      throws IOException {
    return lockAdmin.admin_lock(new AdminLockInfo(serviceId, owner, lockId));    
  }

  
  public boolean release(String serviceId, String lockId) throws IOException{
    return lockAdmin.admin_release(new AdminLockInfo(serviceId, userInfo
        .getHostName(), lockId));
  }
  
  
  public boolean releaseByOwner(String owner) throws IOException{
    return lockAdmin.admin_releaseByOwner(owner);
  }  
  
  
  public boolean releaseByService(String serviceId) throws IOException{
    return lockAdmin.admin_releaseByService(serviceId);
  }   

  
  public boolean releaseAll() throws IOException{
    return lockAdmin.admin_releaseAll();
  }  
  
  
  public boolean format()  throws IOException {
    return lockAdmin.admin_format();
  }
  
  
  public void close() throws IOException{
    
  }
  
  public void run() {
  }

  
  public static void main(String args[]) {
    try {
      LockAdminClient client = new LockAdminClient(new PleiadesConfiguration());
      Thread thread = new Thread(client);
      thread.setDaemon(true);
      thread.start();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }


}
